package ds_industry_2025.ds.YangJuan_2024.T3

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

/*

15、根据dwd或者dws层的数据，请计算每个省份累计订单量，然后根据每个省份订单量从高到低排列，将结果打印到控制台
（使用spark中的show算子，同时需要显示列名）；
例如：可以考虑首先生成类似的临时表A：
province_name	Amount（订单量）
A省	10122
B省	301
C省	2333333

然后生成结果类似如下：其中C省销量最高，排在第一列，A省次之，以此类推。

C省	A省	B省
23333331	10122	301

 */
object t19 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t2")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._

    spark.table("dwd.fact_order_info")
      .where("etl_date=(select max(etl_date) from dwd.fact_order_info)")
      .createOrReplaceTempView("order_info")

    spark.table("dwd.dim_province")
      .where("etl_date=(select max(etl_date) from dwd.dim_province)")
      .createOrReplaceTempView("province")

    val r1 = spark.sql(
      """
        |select
        |province_name,
        |count
        |from(
        |select distinct
        |o.province_id,
        |p.name as province_name,
        |count(*) over(partition by o.province_id,p.name) as count
        |from order_info as o
        |join province as p
        |on p.id=o.province_id
        |) as r1
        |""".stripMargin).orderBy(desc("count"))

    //  todo 拿到所有的省份名称
    val province_arr:Array[String] = r1.select("province_name")
      .map(_(0).toString)
      .collect()


/*
    关键点：pivot 和 max 的配合
    pivot 的作用：1.pivot 会将 province_name 列的值（即省份名称）作为新的列名。
                 2.对于每个省份，pivot 会将其对应的 count 值填充到新列中。
 */
    r1.groupBy()
      .pivot(col("province_name"),province_arr)
      .max("count")
      .show




    spark.close()



  }

}
